package com.kafka.consumer.v2;


import com.kafka.config.KafkaListener;
import com.kafka.service.carMsg.impl.CarMsgServer;
import com.kafka.service.carMsg.impl.SendMsgWebsocket;
import com.kafka.service.carMsg.impl.TestMsg;
import com.kafka.service.strategy.FactorStrategy;
import com.kafka.service.strategy.Handel;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

/**
 * @descriptions: kafka消费监听
 * @author: yunqiang
 * @date: 2022/1/5 12:37
 */
@Component
public class KafkaListenerV2 implements KafkaListener {



    @Override
    public String topic() {
        return "tt";
    }

    @Override
    public void onMessage(ConsumerRecord msg, Consumer consumer) {
        String key = String.valueOf(msg.key());
        String value = String.valueOf(msg.value());
        System.out.println("接收到的key为： "+key);
        System.out.println("接收到的value为： "+value);
        //策略模式
        Handel strategy = FactorStrategy.getInvokeStrategy(key);
        strategy.distributionCarMsg(value);
        //观察者模式
        CarMsgServer carMsgServer = new CarMsgServer();
        carMsgServer.registerObserver(new SendMsgWebsocket());
        carMsgServer.registerObserver(new TestMsg());
        carMsgServer.setInfomation(value);
    }

    @Override
    public void onMessage(Object o) {

    }
}
